package org.sean.framework.kafka.producer;

import org.apache.kafka.clients.producer.ProducerRecord;
import org.sean.framework.code.StatusInfo;
import org.sean.framework.context.SpringApplicationContext;
import org.sean.framework.exception.StatusException;
import org.sean.framework.logging.Logger;
import org.sean.framework.util.IPUtil;
import org.sean.framework.util.ObjectUtil;
import org.sean.framework.util.StringUtil;
import org.sean.framework.web.holder.RequestHolder;
import org.springframework.cloud.commons.util.InetUtils;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * Kafka 生产者
 *
 * @param <V> V
 */
public abstract class BaseProducer<V> {
    // 发送方地址
    public static final String HEADER_PRODUCER_HOST_PORT = "PRODUCER_HOST_PORT";
    private Logger logger = Logger.newInstance(this.getClass());
    private String hostPort = null;

    private KafkaTemplate<Object, Object> kafkaTemplate;

    /**
     * Kafka server 不可用时,发送消息会阻塞工作线程
     */
    private ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 1, 1,
            TimeUnit.MINUTES, new ArrayBlockingQueue<>(10000), new ThreadPoolExecutor.DiscardPolicy());


    BaseProducer(KafkaTemplate<Object, Object> template) {
        this.kafkaTemplate = template;
    }

    /**
     * 发送消息
     *
     * @param message 消息
     * @return 是否发送成功
     */
    public boolean send(V message) {
        return send(null, null, message, false, null);
    }

    /**
     * 发送消息
     *
     * @param key     key
     * @param message 消息
     * @return 是否发送成功
     */
    public boolean send(String key, V message) {
        return send(null, key, message, false, null);
    }

    /**
     * 发送消息
     *
     * @param topic   topic
     * @param key     key
     * @param message 消息
     * @return 是否发送成功
     */
    public boolean send(String topic, String key, V message) {
        return send(topic, key, message, false, null);
    }

    /**
     * 发送消息(同步)
     *
     * @param topic    topic
     * @param key      key
     * @param message  消息体
     * @param callback 回调函数
     * @return 是否发送成功
     */
    public boolean sendSync(String topic, String key, V message,
                            ListenableFutureCallback<SendResult<Object, Object>> callback) {
        return send(topic, key, message, true, callback);
    }

    /**
     * 发送消息
     *
     * @param topic    topic
     * @param key      key
     * @param message  消息体
     * @param isSync   是否同步
     * @param callback 回调函数
     * @return 是否发送成功
     */
    public boolean send(String topic, String key, V message, boolean isSync,
                        ListenableFutureCallback<SendResult<Object, Object>> callback) {
        return send(topic, key, message, null, isSync, callback);
    }

    /**
     * 发送消息
     *
     * @param topic    topic
     * @param key      key
     * @param message  消息体
     * @param headers  头信息
     * @param isSync   是否同步
     * @param callback 回调函数
     * @return 是否发送成功
     */
    public boolean send(String topic, String key, V message, Map<String, String> headers, boolean isSync,
                        ListenableFutureCallback<SendResult<Object, Object>> callback) {
        if (message == null) {
            return false;
        }
        try {
            int flag = 0xFF;
            if (StringUtil.isEmpty(topic)) {
                flag = flag & 0xFB;
            }
            if (key == null) {
                flag = flag & 0xFD;
            }
            ProducerRecord<Object, Object> record;
            switch (flag) {
                case 0xFF:
                    record = new ProducerRecord<>(topic, key, convertMessage(message));
                    break;
                case 0xFB:
                    record = new ProducerRecord<>(kafkaTemplate.getDefaultTopic(), key, convertMessage(message));
                    break;
                case 0xF9:
                    record = new ProducerRecord<>(kafkaTemplate.getDefaultTopic(), convertMessage(message));
                    break;
                case 0xFD:
                    record = new ProducerRecord<>(topic, convertMessage(message));
                    break;
                default:
                    throw new StatusException(StatusInfo.notSupported());
            }
            return send(record, headers, callback, isSync);
        } catch (Exception e) {
            logger.debugThrowable(() -> e);
            return false;
        }
    }

    /**
     * 发送消息
     *
     * @param record   消息
     * @param headers  头信息
     * @param callback 回调
     * @param isSync   是否同步
     * @return 是否发送成功
     */
    private boolean send(ProducerRecord<Object, Object> record, Map<String, String> headers,
                         ListenableFutureCallback<SendResult<Object, Object>> callback, boolean isSync) {
        String errorMsg = StringUtil.formatContent("[KAFKA_SEND_ERROR] message send error : {0} - {1} - {2}",
                record.topic(), record.key(), record.value());
        try {
            if (StringUtil.isEmpty(record.topic()) || ObjectUtil.isEmpty(record.value())) {
                return false;
            }
            if (ObjectUtil.isNotEmpty(headers)) {
                headers.remove("TX_XID");
                headers.remove("tx_xid");
                headers.forEach((k, v) -> record.headers().add(k, v.getBytes()));
            }

            if (StringUtil.isNotEmpty(getHostPort())) {
                record.headers().add(HEADER_PRODUCER_HOST_PORT, hostPort.getBytes());
            }
            try {
                Map<String, String> httpHeaders = RequestHolder.getHeaders();
                if (ObjectUtil.isNotEmpty(httpHeaders)) {
                    httpHeaders.forEach((k, v) -> record.headers().add(k, v.getBytes()));
                }
            } catch (Exception e) {
                // 启动时,各种原因发送消息时,无Header信息
            }

            Future<ListenableFuture<SendResult<Object, Object>>> taskFuture = executor.submit(() -> {
                try {
                    ListenableFuture<SendResult<Object, Object>> future = kafkaTemplate.send(record);
                    ListenableFutureCallback<SendResult<Object, Object>> dc = new ListenableFutureCallback<SendResult<Object, Object>>() {
                        @Override
                        public void onSuccess(SendResult<Object, Object> result) {
                            logger.debug("kafka topic:{} - result:{}-{}",
                                    result.getRecordMetadata().topic(), result.getRecordMetadata(),
                                    result.getProducerRecord());
                            if (callback != null) {
                                callback.onSuccess(result);
                            }
                        }

                        @Override
                        public void onFailure(Throwable ex) {
                            logger.printStackTrace(ex);
                            logger.error(errorMsg);
                            if (callback != null) {
                                callback.onFailure(ex);
                            }
                        }
                    };
                    future.addCallback(dc);
                    return future;
                } catch (Exception e) {
                    logger.error(errorMsg);
                    return null;
                }
            });
            // 同步等待
            if (isSync) {
                SendResult<Object, Object> sendResult = taskFuture.get().get();
                return sendResult != null;
            }
            return true;
        } catch (Exception e) {
            logger.printStackTrace(e);
            logger.error(errorMsg);
            return false;
        }
    }

    private String getHostPort() {
        try {
            if (hostPort == null) {
                InetUtils inetUtils = SpringApplicationContext.getBean(InetUtils.class);
                Integer port = IPUtil.getServerPort();
                hostPort = inetUtils.findFirstNonLoopbackHostInfo().getIpAddress() + ":" + port;
            }
        } catch (Exception e) {
            // do nothing
            hostPort = "";
        }
        return hostPort;
    }

    /**
     * 消息转换
     *
     * @param message 消息
     * @return 转换后消息
     */
    abstract String convertMessage(V message);
}
